# 如果还没有安装 uv，现在就装（这是 2026 年的标准）
curl -LsSf https://astral.sh/uv/install.sh | sh

# 一条命令创建项目和虚拟环境
uv init weather-mcp && cd weather-mcp
uv venv && source .venv/bin/activate

# 安装依赖
uv add "mcp[cli]" httpx python-dotenv

from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP

# 使用服务器元数据初始化
mcp = FastMCP(
    name="weather-intelligence",
    version="1.0.0",
    description="Production weather alerts and risk assessment"
)

# 带安全头的常量
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-mcp/1.0 (your-email@domain.com)"

async def nws_get(endpoint: str) -> dict[str, Any] | None:
    """带正确错误处理的安全 NWS API 客户端"""
    url = f"{NWS_API_BASE}{endpoint}"
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/geo+json"
    }

    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.get(url, headers=headers)
            response.raise_for_status()
            return response.json()
    except httpx.HTTPStatusError as e:
        # 生产环境中应正确记录日志
        print(f"NWS API error: {e.response.status_code}")
        return None
    except Exception:
        # 不要向客户端暴露内部错误
        return None

@mcp.tool()
async def get_alerts(
    state: str = "CA",
    severity: str = "severe"
) -> str:
    """获取某州的活跃天气警报。

    Args:
        state: 两位州代码（如 CA、NY、TX）
        severity: 警报严重级别（severe、moderate、minor）
    """
    # 输入验证
    if len(state) != 2 or not state.isalpha():
        return "Error: State must be a 2-letter code"

    endpoint = f"/alerts/active?area={state.upper()}"
    data = await nws_get(endpoint)

    if not data or "features" not in data:
        return f"No alerts available for {state}"

    alerts = data["features"]
    if not alerts:
        return f"No active alerts for {state}"

    # 为 LLM 消费格式化响应
    formatted = []
    for alert in alerts[:5]:  # 限制为前 5 条
        props = alert["properties"]
        formatted.append(
            f"Event: {props.get('event', 'Unknown')}\n"
            f"Area: {props.get('areaDesc', 'Unknown')}\n"
            f"Severity: {props.get('severity', 'Unknown')}\n"
            f"Description: {props.get('description', 'N/A')[:200]}..."
        )

    return "\n\n".join(formatted)

@mcp.tool()
async def assess_weather_risk(
    location: str,
    days_ahead: int = 3
) -> dict[str, Any]:
    """评估某地未来 N 天的天气风险。

    结合预报、警报和历史模式。
    """
    if days_ahead < 1 or days_ahead > 7:
        return {"error": "days_ahead must be between 1-7"}

    # 获取预报
    # 注意：生产环境中应先对位置进行地理编码
    forecast_data = await nws_get(
        f"/gridpoints/LOX/152,46/forecast"
    )

    if not forecast_data:
        return {"error": "Unable to retrieve forecast"}

    periods = forecast_data["properties"]["periods"][:days_ahead*2]

    # 计算风险指标
    risk_factors = []
    high_temp_days = 0
    precip_days = 0

    for period in periods:
        temp = period.get("temperature", 0)
        if temp > 95:
            high_temp_days += 1
            risk_factors.append(f"High temp: {temp}°F on {period['name']}")

        if period.get("probabilityOfPrecipitation", {}).get("value", 0) > 50:
            precip_days += 1

    risk_score = min(100, (high_temp_days * 20 + precip_days * 15))

    return {
        "location": location,
        "risk_score": risk_score,
        "risk_level": "HIGH" if risk_score > 60 else "MEDIUM" if risk_score > 30 else "LOW",
        "factors": risk_factors,
        "summary": f"{high_temp_days} high-temp days, {precip_days} precipitation days in next {days_ahead} days"
    }

def main():
    """使用 stdio 传输运行 MCP 服务器"""
    mcp.run(transport="stdio")

if __name__ == "__main__":
    main()

# NWS API 标识所需
NWS_USER_AGENT=weather-mcp/1.0 (your-email@domain.com)

# 可选：API 速率限制
CACHE_TTL=300
MAX_REQUESTS_PER_MINUTE=10

code ~/Library/Application\ Support/Claude/claude_desktop_config.json

{
  "mcpServers": {
    "weather-intelligence": {
      "command": "uv",
      "args": [
        "run",
        "--with",
        "mcp[cli]",
        "--with",
        "httpx",
        "python",
        "/absolute/path/to/weather.py"
      ],
      "env": {
        "NWS_USER_AGENT": "weather-mcp/1.0"
      }
    }
  }
}

npx @modelcontextprotocol/inspector uv run weather.py

from functools import wraps
import time

class RateLimiter:
    def __init__(self, max_calls: int, period: float):
        self.max_calls = max_calls
        self.period = period
        self.calls = []

    def __call__(self, func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            now = time.time()
            self.calls = [c for c in self.calls if now - c < self.period]
            if len(self.calls) >= self.max_calls:
                return "Error: Rate limit exceeded. Please wait before retrying."
            self.calls.append(now)
            return await func(*args, **kwargs)
        return wrapper

# 应用到你的工具
@RateLimiter(max_calls=5, period=60.0)
@mcp.tool()
async def get_alerts(...):

FROM python:3.11-slim

WORKDIR /app
COPY weather.py .
COPY .env .

RUN pip install mcp[cli] httpx python-dotenv

CMD ["python", "weather.py"]

# 更改这一行
mcp.run(transport="sse", host="0.0.0.0", port=8000)

# 在你的 Lambda 处理程序中
def lambda_handler(event, context):
    # 从事件体解析 JSON-RPC
    # 路由到 MCP 工具
    # 返回响应
    pass